package qy.jalgotrade.barfeed;

import java.time.ZonedDateTime;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import qy.jalgotrade.bar.Bar;
import qy.jalgotrade.bar.Bar.Frequency;
import qy.jalgotrade.bar.Bars;
import qy.jalgotrade.utils.CommonUtils;
import qy.jalgotrade.utils.PySpecs.ReadOnlyMap;

/**
 * <pre>
 * A non real-time BarFeed responsible for:
 * - Holding bars in memory.
 * - Aligning them with respect to time.
 * 
 * Subclasses should:
 * - Forward the call to start() if they override it.
 * </pre>
 * 
 * @author qy
 *
 */
public abstract class MemBarFeed extends BaseBarFeed {

	private final Map<String, List<Bar>> __bars;

	private Map<String, Integer> __nextPos;

	private Map<String, Integer> __nextPosDispatching;

	private boolean __started;

	private ZonedDateTime __currDateTime;

	private ZonedDateTime __currDateTimeDispatching;

	/**
	 * @param frequency frequency
	 * @throws Exception Exception
	 */
	public MemBarFeed(Frequency frequency) throws Exception {

		this(frequency, 0);
	}

	/**
	 * @param frequency frequency
	 * @param maxLen maxLen
	 * @throws Exception Exception
	 */
	public MemBarFeed(Frequency frequency, int maxLen) throws Exception {

		super(frequency, maxLen);

		__bars = new HashMap<>();
		__nextPos = new HashMap<>();
		__nextPosDispatching = null;
		__started = false;
		__currDateTime = null;
		__currDateTimeDispatching = null;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see qy.jalgotrade.barfeed.BaseBarFeed#getCurrentDateTime()
	 */
	@Override
	public ZonedDateTime getCurrentDateTime() {

		if (isDispatching()) {
			return __currDateTimeDispatching;
		} else {
			return __currDateTime;
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see qy.jalgotrade.barfeed.BaseBarFeed#getNextBars()
	 */
	@Override
	@SuppressWarnings("unchecked")
	public Bars getNextBars() throws Exception {

		// All bars must have the same datetime. We will return all the ones with the smallest datetime.
		ZonedDateTime smallestDateTime = peekDateTime();
		if (smallestDateTime == null) {
			return null;
		}

		__nextPosDispatching = (Map<String, Integer>) ((HashMap<String, Integer>) __nextPos).clone();
		// Make a second pass to get all the bars that had the smallest datetime.
		Map<String, Bar> ret = new HashMap<>();
		for (Entry<String, List<Bar>> e : __bars.entrySet()) {
			String instrument = e.getKey();
			List<Bar> bars = e.getValue();
			int nextPos = __nextPos.get(instrument);
			if (nextPos < bars.size() && bars.get(nextPos).getDateTime().equals(smallestDateTime)) {
				ret.put(instrument, bars.get(nextPos));
				__nextPos.put(instrument, __nextPos.get(instrument) + 1);
			}
		}

		if (__currDateTime != null && __currDateTime.equals(smallestDateTime)) {
			throw new Exception(String.format("Duplicate bars found for %s on %s", ret.keySet(), smallestDateTime));
		}

		__currDateTimeDispatching = __currDateTime;

		__currDateTime = smallestDateTime;
		return new Bars(ret);
	}

	/**
	 * 
	 * @param instrument instrument
	 * @param bars bars
	 * @throws Exception Exception
	 */
	public void addBarsFromSequence(String instrument, List<Bar> bars) throws Exception {

		if (__started) {
			throw new Exception("Can't add more bars once you started consuming bars");
		}

		if (!__bars.containsKey(instrument)) {
			__bars.put(instrument, new ArrayList<>());
		}
		if (!__nextPos.containsKey(instrument)) {
			__nextPos.put(instrument, 0);
		}

		// Add and sort the bars
		__bars.get(instrument).addAll(bars);
		Collections.sort(__bars.get(instrument), new Comparator<Bar>() {

			@Override
			public int compare(Bar a, Bar b) {

				return a.getDateTime().compareTo(b.getDateTime());
			}
		});

		registerInstrument(instrument);
	}

	/**
	 * 
	 */
	public void loadAll() {

		__iter__().forEachRemaining(params -> {
			@SuppressWarnings("unused")
			ZonedDateTime dateTime = params.getLeft();
			@SuppressWarnings("unused")
			ReadOnlyMap<String, Bar> bars = params.getRight();
		});
	}

	/**
	 * 
	 */
	@Override
	public void reset() throws Exception {

		__nextPos = new HashMap<>();
		for (String instrument : __bars.keySet()) {
			__nextPos.put(instrument, 0);
		}
		__currDateTime = null;
		__currDateTimeDispatching = null;
		super.reset();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see qy.jalgotrade.event.Subject#start()
	 */
	@Override
	public void start() {

		__started = false;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see qy.jalgotrade.event.Subject#stop()
	 */
	@Override
	public void stop() {

	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see qy.jalgotrade.event.Subject#join()
	 */
	@Override
	public void join() {

	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see qy.jalgotrade.event.Subject#eof()
	 */
	@Override
	public boolean eof() {

		boolean ret = true;
		// Check if there is at least one more bar to return.
		for (Entry<String, List<Bar>> e : __bars.entrySet()) {

			String instrument = e.getKey();
			List<Bar> bars = e.getValue();

			int nextPos;
			if (isDispatching()) {
				nextPos = __nextPosDispatching.get(instrument);
			} else {
				nextPos = __nextPos.get(instrument);
			}

			if (nextPos < bars.size()) {
				ret = false;
				break;
			}
		}
		return ret;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see qy.jalgotrade.event.Subject#peekDateTime()
	 */
	@Override
	public ZonedDateTime peekDateTime() {

		ZonedDateTime ret = null;

		for (Entry<String, List<Bar>> e : __bars.entrySet()) {

			String instrument = e.getKey();
			List<Bar> bars = e.getValue();

			int nextPos;
			if (isDispatching()) {
				nextPos = __nextPosDispatching.get(instrument);
			} else {
				nextPos = __nextPos.get(instrument);
			}

			if (nextPos < bars.size()) {
				ret = (ZonedDateTime) CommonUtils.safeMin(ret, bars.get(nextPos).getDateTime()); // TODO:
			}
		}
		return ret;
	}
}
